package dataframe

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Row, SparkSession}

object DataFrame_CreateTest {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local[*]")
    conf.setAppName("DataFrame_CreateTest")

    val spark = SparkSession
      .builder()
      .config(conf)
      .getOrCreate()

    // 创建单列DataFrame，默认列名是id，类型是LongType
    val df: DataFrame = spark.range(5).toDF()
    df.show()

    val df1: DataFrame = spark.range(10).toDF("number")
    df1.show()

    val df2: DataFrame = spark.range(5, 10).toDF("num")
    val df3: DataFrame = spark.range(5, 15, 2).toDF("num2")
    df2.show()
    df3.show()

    // 将一个元组集合转换为一个DataFrame，可创建多列DataFrame
    // 导入隐式转换，如将rdd转换为DataFrame或者将元组集合转换为DataFrame
    import spark.implicits._

    val moviesDF: DataFrame = Seq(
      ("马特·达蒙", "谍影重重:极限伯恩", 2007L), ("马特·达蒙", "心灵捕手", 1997L)
    ).toDF("演员","电影","年份")
    // 输出模式
    moviesDF.printSchema()
    // 显示数据
    moviesDF.show()


    // 构造DataFrame-从RDD创建
    // 使用包含Row数据(以元组的形式)的RDD创建DataFrame
    val personDF: DataFrame = spark.sparkContext.parallelize(List(
      ("张三", 23), ("李四", 18), ("王老五", 35)
    )).toDF("name", "age")
    personDF.printSchema()
    personDF.show()
    println("*"*100)

    // 使用case class和反射来创建DataFrame
    val personDF1: DataFrame = spark.sparkContext.parallelize(List(
      Person("张三", 29),
      Person("李四", 30),
      Person("王老五", 19)
    )).toDF()

    personDF1.printSchema()
    personDF1.show()

    val personDF2: DataFrame = List(
      Person("张三", 29),
      Person("李四", 30),
      Person("王老五", 19)
    ).toDF()
    personDF2.printSchema()
    personDF2.show()

    // 如果上面示例中的集合元素不是Person对象，而是字符串的话，那么需要我们自己通过转换来构造出Person集合来。
    val strPersonList: List[String] = List("张三,29", "李四,30", "王老五,19")
    val personDF3: DataFrame = spark
      .sparkContext
      .parallelize(strPersonList)
      .map(str => {
        val fields = str.split(",")
        Person(fields(0).trim, fields(1).trim.toLong)
      }).toDF()
    personDF3.printSchema()
    personDF3.show()
    println("*"*100)


    // 使用指定的模式创建DataFrame
    // 通过一个编程接口，先构造出一个特定的模式（schema），然后使用该模式（schema）创建一个DataFrame。
    // 这需要使用SparkSession的方法createDataFrame来创建

    // 构造一个RDD
    val peopleRDD: RDD[Row] = spark.sparkContext.parallelize(
      Seq(
        Row("张三", 30),
        Row("李四", 25),
        Row("王老五", 35)
      )
    )
    // 指定一个Schema(模式)
    val schema: StructType = new StructType(Array(
      new StructField("name", StringType, false),
      new StructField("age", IntegerType, false)
    ))
    // 从给定的RDD应用给定的Schema创建一个DataFrame
    val peopleDF: DataFrame = spark.createDataFrame(peopleRDD, schema)
    // 查看DataFrame Schema
    peopleDF.printSchema()
    // 输出查看数据
    peopleDF.show()




    spark.stop()
  }

  case class Person(name:String,age:Long)
}
